/**
 * file: TaskDispatcher.java
 */

package pcsd.server;

import java.rmi.RemoteException;
import java.util.Deque;
import java.io.FileWriter;
import java.io.IOException;

import pcsd.worker.JobProcessingWorker;
import pcsd.server.Monitor;
import pcsd.Result;
import pcsd.Task;

/**
 * A runnable that dispatches tasks from a given worker queue to a given
 * worker.
 * 
 * @author vmarcos, Tim van Deurzen, Benjamin Ertl, Justinas Murzinas, Boyan Mihaylov
 */
public class TaskDispatcher implements Runnable {
	
	/**
	 * The result file name prefix.
	 */
	public static final String FILE_NAME_PREFIX = "task-result-";
	
	/**
	 * The threshold showing after how many misses
	 * the thread should go to sleep
	 */
	public static final Integer THREAD_SLEEP_THRESHOLD = 5;
	
	/** The task queue. */
	private Deque<Task> workerQueue;
	
	/** The RMI stub for the worker. */
	private JobProcessingWorker worker;
	
	/** Worker identifier. */
	private int workerNum;

    /** Monitor used for signaling our parent thread. */
	private Monitor monitor;

    /** Log file. */
	private FileWriter log;
	
	/** Counts how many times the thread has done no work. */
	private int missTimes;
	
	/**
	 * Instantiates the task dispatcher.
	 */
	public TaskDispatcher( Deque<Task> workerQueue
	                     , JobProcessingWorker worker
	                     , int workerNum
	                     , Monitor monitor) throws RemoteException {

		this.workerQueue = workerQueue;
		this.worker = worker;
		this.workerNum = workerNum;
		this.monitor = monitor;
		this.missTimes = 0;
        
        try {
            /* We open the file set the append flag to true. */
            this.log = new FileWriter(FILE_NAME_PREFIX + workerNum, true);
        } catch (IOException e) {
            throw new RemoteException("Couldn't open log file!", e);
        }
	}
	
	/**
	 * Main code for the thread.
	 *
     * The double try-catch system for InterruptedException is due to the fact
     * that run is not allowed to throw InterruptedException due to interface
     * restrictions.
	 */
	@Override
	public void run() {
		Task task = null;
		
		// The worker is active until it is shut down externally
		while (true) {
			// Synchronize the access to the worker queue, so that
			// another thread cannot modify it meanwhile (i.e. add new tasks)
			synchronized (this.workerQueue) {
				if (!this.workerQueue.isEmpty()) {
					// Executing many tasks continuously can prevent the
					// scheduler from adding new tasks. Thus, we get
					// one task and then free the lock on the queue
					task = this.workerQueue.pop();
			    	
			    	// reset the counter
			    	this.missTimes = 0;
				}
				else {
					// No work is done because the queue is empty
					this.missTimes++;
					if (this.missTimes == Integer.MAX_VALUE) {
						this.missTimes = THREAD_SLEEP_THRESHOLD;
					}
				}
			}
			
			// Execute the task out of the sync block. This way we let
			// other threads to use the queue.
			if (task != null) {
				this.executeTask(task);
				task = null;
			}
			
			// update monitor
			synchronized (this.monitor) {
				this.monitor.workerNum = workerNum;
            	
				// Notify any thread waiting for the queue to become empty
				// Here we do a double check. This may safe some time to acquire
				// a lock if the queue is empty
				if (this.workerQueue.isEmpty()) {
					synchronized (this.workerQueue) {
		        		if (this.workerQueue.isEmpty()) {
		        			this.monitor.notify();
		    			}
					}
				}
			}
			
			// Sleep for a while so that other threads can also work
			this.sleep();
		}
	}

	/**
	 * Executes a task. This method is just a wrapper over the worker.
	 * The result is written to a file if there is no errors.
	 * @param task
	 */
    public void executeTask(Task task) {
	    try {
	    	Result result = worker.executeTask(task);
	    	if (result.status == 0) {
	    		this.writeToFile(result);
	    	}
	    	else {
	    		System.out.println("Worker " + this.workerNum + ": Task returned " 
	    				+ result.status);
	    	}
	    }
	    catch (RemoteException re) {
	    	System.out.println("Worker " + this.workerNum + 
	    			": A remote exception occurred");
	    	re.printStackTrace();
	    }
	    catch (Exception e) {
	    	System.out.println("Worker " + this.workerNum + 
	    			": An unexpected exception occurred");
	    	e.printStackTrace();
	    }
    }
    
    /**
     * Depending on how many time the thread has done no work 
     * we may simply yield its executing time or sleep. If there is 
     * no work for some longer time, it's better to sleep than 
     * just switching thread contexts
     */
    private void sleep() {
    	if (this.missTimes < THREAD_SLEEP_THRESHOLD) {
			Thread.yield();
		}
		else {
			long timeToSleep;
			if (this.missTimes < 50)
				timeToSleep = 100 * this.missTimes;
			else
				timeToSleep = 5000L;
			
			try {
				Thread.sleep(timeToSleep);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
    }

	/**
	 * Writes to the worker file.
	 */
    private void writeToFile(Result result) throws RemoteException {
	    try {
            /** 
             * Log the result to the worker file also logging the time in
             * milisections.
             */
            this.log.write( "[" +  System.currentTimeMillis() + "] "
                    + "Worker " 
                    + workerNum 
                    + " produced result value "
                    + result.output
                    + "\n");
            this.log.flush();
        } catch(IOException e) {
            /** 
             * This is a show-stopping exception.
             */
            throw new RemoteException("Could not write worker log file!", e);
        }
    }
}
